package com.iqiyi.pps.epg.core.utils;

import com.alibaba.fastjson.JSONObject;
import com.iqiyi.kiwi.utils.Configuration;
import com.iqiyi.pps.epg.core.model.channel.ChannelBase;
import com.iqiyi.pps.epg.core.model.program.ProgramData;
import com.iqiyi.pps.epg.core.model.rtmp.RtmpServer;
import com.iqiyi.pps.epg.core.model.stream.LiveChannelStream;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.program.ProgramService;
import com.iqiyi.pps.epg.core.service.rtmp.RtmpService;
import com.iqiyi.pps.epg.core.service.stream.LiveChannelStreamService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by karkarrotxu on 2015/11/2.
 */
public final class ChannelNotifyUtil {
    private static Logger logger = LoggerFactory.getLogger(ChannelNotifyUtil.class);
    private static ChannelService channelService = null;
    private static ProgramService programService = null;
    private static LiveChannelStreamService liveChannelStreamService = null;
    private static RtmpService rtmpService = null;
    public static final String CHANNEL_NOTIFY_URL = Configuration.getString("channel.notify.url");
    public static final long SLEEP_TIME = 1500;
    private final static BlockingQueue<NotifyTask> tasks = new LinkedBlockingQueue<NotifyTask>();
    private final static int WARN_LIMIT = 1000;
    private static final long CHECK_SIZE_GAP_TIME = 300000;

    private static Thread taskThread = null;

    private ChannelNotifyUtil() {
    }

    private static ChannelService getChannelService() {
        if (channelService == null) {
            channelService = (ChannelService) ApplicationContextUtil.getBean(ChannelService.class);
        }
        return channelService;
    }

    static {
        taskThread = new Thread(new Runnable() {
            private long checkSizeTime = System.currentTimeMillis();
            @Override
            public void run() {
                while (true) {
                    try {
                        if (System.currentTimeMillis() > checkSizeTime + CHECK_SIZE_GAP_TIME) {
                            checkSizeTime = System.currentTimeMillis();
                            int currentSize = tasks.size();
                            if (currentSize > WARN_LIMIT) {
                                logger.warn("BlockingQueue task over {},current is {}", WARN_LIMIT, currentSize);
                            }
                        }
                        NotifyTask task = tasks.take();
                        excuteQuery(task);
                        Thread.sleep(SLEEP_TIME);
                    } catch (InterruptedException e) {
                        logger.error("InterruptedException ", e);
                    } catch (Exception e) {
                        logger.error("exception", e);
                    }
                }
            }
        });
        taskThread.start();
    }


    private static ProgramService getProgramService() {
        if (programService == null) {
            programService = (ProgramService) ApplicationContextUtil.getBean(ProgramService.class);
        }
        return programService;
    }

    private static LiveChannelStreamService getLiveChannelStreamService() {
        if (liveChannelStreamService == null) {
            liveChannelStreamService = (LiveChannelStreamService) ApplicationContextUtil.getBean(LiveChannelStreamService.class);
        }
        return liveChannelStreamService;
    }

    private static RtmpService getRtmpService() {
        if (rtmpService == null) {
            rtmpService = (RtmpService) ApplicationContextUtil.getBean(RtmpService.class);
        }
        return rtmpService;
    }

    public static boolean checkTypeLB(long channelId, int liveType) {
        try {
            if (0 == liveType) {
                ChannelBase channelBase = getChannelService().getChannelBase(channelId);
                if (null == channelBase) {
                    return false;
                }
                liveType = channelBase.getLiveType();
            }
            if (liveType != ChannelBase.LIVE_TYPE_LB) {
                return false;
            }
            return true;
        } catch (Exception e) {
            logger.error("[checkTypeLB]", e);
            return false;
        }
    }

    public static boolean isInList(long channelId, final long programId) {
        return isInList(channelId, new ArrayList<Long>() {
            {
                add(programId);
            }
        });
    }

    public static boolean isInList(long channelId, List<Long> programIdList) {
        try {
            List<ProgramData> currentNextList = getProgramService().getCurrentAndNextProgram(channelId, 15);
            boolean isInList = false;
            if (null == currentNextList || currentNextList.size() == 0) {
                return false;
            }

            for (ProgramData programData : currentNextList) {
                if (programIdList.contains(programData.getId())) {
                    isInList = true;
                    break;
                }
            }
            if (!isInList) {
                return false;
            }
            return true;
        } catch (Exception e) {
            logger.error("[isInList]", e);
            return false;
        }
    }

    public static void checkProgramModification(long channelId, long programId, int liveType) {
        logger.info("[ChannelNotifyUtil][checkProgramModification][start][channelId={}][programId={}][liveType={}]",
                channelId, programId, liveType);
        if (!checkTypeLB(channelId, liveType)) {
            return;
        }
        if (!isInList(channelId, programId)) {
            return;
        }
        forceNotify(channelId);
        logger.info("[ChannelNotifyUtil][checkProgramModification][end][channelId={}][programId={}]", channelId, programId);
    }

    public static void forceNotify(long channelId) {
        try {
            List<LiveChannelStream> streams = getLiveChannelStreamService().getByFatherId(channelId);
            if (null == streams) {
                return;
            }
            for (int i = 0; i < streams.size(); ++i) {
                Map<String, Object> params = new HashMap<>();
                LiveChannelStream stream = streams.get(i);
                params.put("channelid", String.valueOf(channelId));
                params.put("obj", "rtmp://" + getPrivateIpByPublicIp(stream.getStreamIp()) + "/live/" + stream.getStreamName());
                tasks.add(new NotifyTask(channelId, System.currentTimeMillis(), params));
            }
            logger.info("[ChannelNotifyUtil][forceNotify][addTask][channelId={}]", channelId);
        } catch (Exception e) {
            logger.error("[ChannelNotifyUtil][forceNotify][failed]", e);
        }
    }

    public static void excuteQuery(NotifyTask task) {
        JSONObject response = ApiHttpRequestUtil.getMethod(CHANNEL_NOTIFY_URL, task.getParams());
        logger.info("[ChannelNotifyUtil][excuteQuery][channelid={}][obj={}][response=]{}",
                task.getChannelId(), task.getParams().get("obj"), (response == null ? "npe" : response.toJSONString()));

    }

    public static String getPrivateIpByPublicIp(String publicIp) {
        if (null == publicIp || "".equals(publicIp)) {
            return "";
        }
        try {
            RtmpServer rtmpServer = getRtmpService().getByPublicIp(publicIp);
            if (null == rtmpServer || null == rtmpServer.getPrivateIp()) {
                return "";
            }
            return rtmpServer.getPrivateIp();
        } catch (Exception e) {
            logger.info("[ChannelNotifyUtil][getPrivateIpByPublicIp][publicIp={}][failed]", publicIp);
            return "";
        }
    }

    static class NotifyTask {
        public NotifyTask(long channelId, long ct, Map<String, Object> params) {
            this.channelId = channelId;
            this.createTime = ct;
            this.params = params;
        }

        private long channelId;

        private long createTime;

        public Map<String, Object> getParams() {
            return params;
        }

        public void setParams(Map<String, Object> params) {
            this.params = params;
        }

        private Map<String, Object> params;

        public long getCreateTime() {
            return createTime;
        }

        public void setCreateTime(long createTime) {
            this.createTime = createTime;
        }

        public long getChannelId() {
            return channelId;
        }

        public void setChannelId(long channelId) {
            this.channelId = channelId;
        }
    }
}


